from kafka import KafkaConsumer
from crawler import CectCrawler
from crawler import WeirdCrawler
from crawler import crawler
from flask import Flask

app = Flask(__name__)


class CrawlerJobConsumer:
    def __init__(self, kafkahost, kafkaport, kafkatopic, groupid):
        self.kafkaHost = kafkahost
        self.kafkaPort = kafkaport
        self.kafkatopic = kafkatopic
        self.groupid = groupid
        self.consumer = KafkaConsumer(self.kafkatopic, group_id=self.groupid,
                                      bootstrap_servers='{kafka_host}:{kafka_port}'.format(
                                          kafka_host=self.kafkaHost,
                                          kafka_port=self.kafkaPort))

    def consume_data(self):
        msgs = []
        try:
            for message in self.consumer:
                # print json.loads(message.value)
                do_job(str(message.value,encoding="utf-8"))
                print("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition, message.offset, message.key, message.value))
        except KeyboardInterrupt as e:
            print(e)
        return msgs


def do_job(crawler_name):
    if crawler_name == 'cect':
        app.logger.info("CectCrawler被调用")
        CectCrawler.main()
    else:
        app.logger.info("1024 Crawler被调用")
        WeirdCrawler.main()


if __name__ == '__main__':
    consumer = CrawlerJobConsumer('192.168.56.10', 9092, "crawler.queue","crawler_consumer")
    message = consumer.consume_data()